package tk.mynoteapp.dev.demo03.threadpooloom;

import jodd.util.concurrent.ThreadFactoryBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
 * 一、线程池的声明需要手动进行
 * 我同样不建议使用 Executors 提供的两种快捷的线程池，原因如下：
 *
 * - 我们需要根据自己的场景、并发情况来评估线程池的几个核心参数，包括核心线程数、最大线程数、线程回收策略、
 *   工作队列的类型，以及拒绝策略，确保线程池的工作行为符合需求，一般都需要设置有界的工作队列和可控的线程数。
 * - 任何时候，都应该为自定义线程池指定有意义的名称，以方便排查问题。当出现线程数量暴增、线程死锁、线程占用
 *   大量 CPU、线程执行出现异常等问题时，我们往往会抓取线程栈。此时，有意义的线程名称，就可以方便我们定位问题。
 * - 除了建议手动声明线程池以外，我还建议用一些监控手段来观察线程池的状态。线程池这个组件往往会表现得任劳任怨、
 *   默默无闻，除非是出现了拒绝策略，否则压力再大都不会抛出一个异常。如果我们能提前观察到线程池队列的积压，或者
 *   线程数量的快速膨胀，往往可以提早发现并解决问题。
 */
@RestController
@RequestMapping("threadpooloom")
public class ThreadPoolOOMController {
    private Logger logger = LoggerFactory.getLogger(ThreadPoolOOMController.class);

    /**
     * 打印线程池信息
     * @param threadPool
     */
    private void printStats(ThreadPoolExecutor threadPool) {
        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> {
            logger.info("=========================");
            // 线程数
            logger.info("Pool Size: {}", threadPool.getPoolSize());
            // 活跃线程数
            logger.info("Active Threads: {}", threadPool.getActiveCount());
            // 完成了多少任务
            logger.info("Number of Tasks Completed: {}", threadPool.getCompletedTaskCount());
            // 队列中积压的任务数
            logger.info("Number of Tasks in Queue: {}", threadPool.getQueue().size());
            logger.info("=========================");
        }, 0, 1, TimeUnit.SECONDS);
    }

    /**
     * Executors.newFixedThreadPool：工作队列采用了LinkedBlockingQueue 长度为 Integer.MAX_VALUE 的队列，相当于是无界的
     * 原因：任务较多并且执行较慢的话，队列可能会快速积压，撑爆内存导致 OOM
     */
    @GetMapping("oom1")
    public void oom1() throws InterruptedException {
        ThreadPoolExecutor threadPool = (ThreadPoolExecutor) Executors.newFixedThreadPool(1);
        printStats(threadPool);
        // 循环1亿次
        for (int i = 0; i < 100000000; i++) {
            threadPool.execute(() -> {
                String payload = IntStream.rangeClosed(1, 1000000) // 1百万
                        .mapToObj(__ -> "a")
                        .collect(Collectors.joining("")) + UUID.randomUUID().toString();
                try {
                    TimeUnit.HOURS.sleep(1);
                } catch (InterruptedException e) {
                }
                logger.info(payload);
            });
        }
        threadPool.shutdown();
        threadPool.awaitTermination(1, TimeUnit.HOURS);
    }

    /**
     * Executors.newCachedThreadPool：最大线程数设置为了Integer.MAX_VALUE
     * 原因：
     * 1）最大线程数是Integer.MAX_VALUE相当于没有上限
     * 2）工作队列 SynchronousQueue 是一个没有存储空间的阻塞队列
     */
    @GetMapping("oom2")
    public void oom2() throws InterruptedException {
        ThreadPoolExecutor threadPool = (ThreadPoolExecutor) Executors.newCachedThreadPool();
        printStats(threadPool);
        for (int i = 0; i < 100000000; i++) {
            threadPool.execute(() -> {
                String payload = IntStream.rangeClosed(1, 1000000)
                        .mapToObj(__ -> "a")
                        .collect(Collectors.joining("")) + UUID.randomUUID().toString();
                try {
                    TimeUnit.HOURS.sleep(1);
                } catch (InterruptedException e) {
                }
                logger.info(payload);
            });
        }
        threadPool.shutdown();
        threadPool.awaitTermination(1, TimeUnit.HOURS);
    }

    /*
     * 线程池默认的工作行为：
     *   - 不会初始化 corePoolSize 个线程，有任务来了才创建工作线程；
     *   - 当核心线程满了之后不会立即扩容线程池，而是把任务堆积到工作队列中；
     *   - 当工作队列满了后扩容线程池，一直到线程个数达到 maximumPoolSize 为止；
     *   - 如果队列已满且达到了最大线程后还有任务进来，按照拒绝策略处理；
     *   - 当线程数大于核心线程数时，线程等待 keepAliveTime 后还是没有任务需要处理的话，收缩线程到核心线程数。
     *
     * 改变默认行为：
     *   - 声明线程池后立即调用 prestartAllCoreThreads 方法，来启动所有核心线程；
     *   - 传入 true 给 allowCoreThreadTimeOut 方法，来让线程池在空闲的时候同样回收核心线程。
     */

    @GetMapping("right")
    public int right() throws InterruptedException {
        AtomicInteger atomicInteger = new AtomicInteger();
        /*
        * - 不建议使用Executors提供的方法创建线程池
        * - 使用ThreadPoolExecutor显示创建线程池并指定可控的线程数，有界的工作队列、有意义的名称、合理的拒绝策略
        * */
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
                2, 5, // 2个核心线程，5个最大线程
                5, TimeUnit.SECONDS, // 非核心线程数没有任务处理等待5s
                new ArrayBlockingQueue<>(10),// 工作队列容量为10
                new ThreadFactoryBuilder().setNameFormat("demo-threadpool-%d").get(),// 设置线程名称
                new ThreadPoolExecutor.AbortPolicy());// 指定拒绝策略
        printStats(threadPool);
        IntStream.rangeClosed(1, 20).forEach(i -> {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
            }
            int id = atomicInteger.incrementAndGet();
            try {
                threadPool.submit(() -> {
                    logger.info("{} started", id);
                    try {
                        TimeUnit.SECONDS.sleep(10);
                    } catch (InterruptedException e) {
                    }
                    logger.info("{} finished", id);
                });
            } catch (Exception ex) {
                logger.error("error submitting task {}", id, ex);
                atomicInteger.decrementAndGet();
            }

        });
        TimeUnit.SECONDS.sleep(60);
        return atomicInteger.intValue();
    }

    /**
     * 问题：更激进一点，优先开启更多的线程，而把队列当成一个后备方案呢？
     * 思路：
     *   - 由于线程池在工作队列满了无法入队的情况下会扩容线程池，那么我们
     *     是否可以重写队列的 offer 方法，造成这个队列已满的假象呢？
     *   - 由于我们 Hack 了队列，在达到了最大线程后势必会触发拒绝策略，那
     *     么能否实现一个自定义的拒绝策略处理程序，这个时候再把任务真正插入队列呢？
     * @return
     * @throws InterruptedException
     */
    @GetMapping("better")
    public int better() throws InterruptedException {
        //这里开始是激进线程池的实现
        BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>(10) {
            @Override
            public boolean offer(Runnable o) {
                //先返回false，造成队列满的假象，让线程池优先扩容
                return false;
            }
        };

        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
                2, 5,
                5, TimeUnit.SECONDS,
                queue, // 使用hack后的队列，造成队列已满的假象，等到线程达到maximumPoolSize
                new ThreadFactoryBuilder().setNameFormat("demo-threadpool-%d").get(),
                (r, executor) -> {// 自定义拒绝策略，线程数达到maximumPoolSize后会执行拒绝策略，此时将线程放入队列
                    try {
                        //等出现拒绝后再加入队列
                        //如果希望队列满了阻塞线程而不是抛出异常，那么可以注释掉下面三行代码，修改为executor.getQueue().put(r);
                        if (!executor.getQueue().offer(r, 0, TimeUnit.SECONDS)) {
                            throw new RejectedExecutionException("ThreadPool queue full, failed to offer " + r.toString());
                        }
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                }
        );
        //激进线程池实现结束

        // 监控线程池
        printStats(threadPool);

        //任务编号计数器
        AtomicInteger atomicInteger = new AtomicInteger();

        //每秒提交一个任务，每个任务耗时10秒执行完成，一共提交20个任务
        IntStream.rangeClosed(1, 20).forEach( i -> {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            int id = atomicInteger.incrementAndGet();
            try {
                threadPool.submit(() -> {
                    logger.info("{} started", id);
                    try {
                        TimeUnit.SECONDS.sleep(10);
                    } catch (InterruptedException e) {
                    }
                    logger.info("{} finished", id);
                });
            } catch (Exception ex) {
                logger.error("error submitting task {}", id, ex);
                atomicInteger.decrementAndGet();
            }
        });
        TimeUnit.SECONDS.sleep(60);
        return atomicInteger.intValue();
    }
}
